package com.okex.demo.service;

import com.okex.demo.config.WebSocketEventListener;
import com.okex.demo.model.TickerInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.event.EventListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 数据轮询服务
 * 按需执行数据查询和推送，只有在有WebSocket连接和产品订阅时才会执行
 * 通过事件机制响应WebSocket连接和订阅事件
 */
@Slf4j
@Service
public class DataPollingService {

    @Autowired
    private OkexApiService okexApiService;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Value("${okex.default.inst-id:BTC-USDT}")
    private String defaultInstId;

    @Value("${okex.polling.interval:2000}")
    private long pollingInterval; // 默认2秒查询一次

    private final AtomicBoolean isRunning = new AtomicBoolean(false);
    private final Set<String> subscribedProducts = ConcurrentHashMap.newKeySet();
    private final Map<String, Long> lastUpdateTimes = new ConcurrentHashMap<>();
    private ScheduledExecutorService executorService;

    @PostConstruct
    public void init() {
        log.info("数据轮询服务初始化");
        // 不立即启动查询任务，等待WebSocket连接建立后再启动
        // 默认添加默认产品
        subscribedProducts.add(defaultInstId);
    }

    @PreDestroy
    public void shutdown() {
        stopPolling();
    }

    /**
     * 监听WebSocket连接事件
     */
    @EventListener
    public void handleWebSocketConnectionEvent(WebSocketEventListener.WebSocketConnectionEvent event) {
        if (event.isConnected()) {
            // 首次连接建立
            startPolling();
        } else {
            // 最后一个连接断开
            stopPolling();
        }
    }

    /**
     * 监听产品订阅事件
     */
    @EventListener
    public void handleProductSubscriptionEvent(WebSocketEventListener.ProductSubscriptionEvent event) {
        String productId = event.getProductId();
        if (event.isSubscribed()) {
            // 产品被订阅
            log.info("添加产品订阅到轮询列表: {}", productId);
            subscribedProducts.add(productId);

            // 如果服务在运行，立即请求一次该产品的数据
            if (isRunning.get()) {
                queryProductData(productId);
            }
        } else {
            // 产品取消订阅
            log.info("从轮询列表移除产品订阅: {}", productId);
            // 不要移除默认产品
            if (!productId.equals(defaultInstId)) {
                subscribedProducts.remove(productId);
            }
        }
    }

    /**
     * 启动数据轮询
     * 只有在首次WebSocket连接建立时调用此方法
     */
    public synchronized void startPolling() {
        if (isRunning.get()) {
            log.debug("数据轮询任务已经在运行中");
            return;
        }

        log.info("启动数据轮询任务，间隔: {}毫秒", pollingInterval);
        executorService = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, "data-polling-thread");
            thread.setDaemon(true);
            return thread;
        });
        executorService.scheduleAtFixedRate(this::pollData, 0, pollingInterval, TimeUnit.MILLISECONDS);
        isRunning.set(true);
    }

    /**
     * 停止数据轮询
     * 在最后一个WebSocket连接断开时调用此方法
     */
    public synchronized void stopPolling() {
        if (!isRunning.get()) {
            return;
        }

        log.info("停止数据轮询任务");
        if (executorService != null) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        isRunning.set(false);
    }

    /**
     * 按需执行数据查询和推送
     * 只查询和推送有订阅者的产品
     */
    private void pollData() {
        try {
            // 确保订阅列表至少包含默认产品
            if (subscribedProducts.isEmpty()) {
                subscribedProducts.add(defaultInstId);
            }

            log.debug("执行数据查询: 有 {} 个订阅的产品", subscribedProducts.size());

            // 复制一份集合，避免并发修改问题
            Set<String> productsToPoll = new HashSet<>(subscribedProducts);

            // 只查询和推送订阅的产品
            for (String instId : productsToPoll) {
                queryProductData(instId);
            }
        } catch (Exception e) {
            log.error("数据轮询任务执行异常: {}", e.getMessage(), e);
        }
    }

    /**
     * 查询单个产品数据并推送
     */
    private void queryProductData(String instId) {
        try {
            // 获取最新行情
            TickerInfo ticker = okexApiService.getTicker(instId);
            if (ticker != null) {
                // 设置更新时间
                Date now = new Date();
                ticker.setUpdateTime(now);
                lastUpdateTimes.put(instId, now.getTime());

                // 推送到特定产品主题
                String specificTopic = "/topic/ticker/" + instId;
                log.debug("推送产品行情: {} 到 {}", instId, specificTopic);
                messagingTemplate.convertAndSend(specificTopic, ticker);

                // 如果是默认产品，同时推送到通用主题
                if (instId.equals(defaultInstId)) {
                    messagingTemplate.convertAndSend("/topic/ticker", ticker);
                }
            }
        } catch (Exception e) {
            log.warn("获取产品 {} 行情数据失败: {}", instId, e.getMessage());
        }
    }

    /**
     * 获取产品最后更新时间
     */
    public long getLastUpdateTime(String instId) {
        return lastUpdateTimes.getOrDefault(instId, 0L);
    }

    /**
     * 判断产品是否需要更新
     * @param instId 产品ID
     * @param threshold 时间阈值（毫秒）
     */
    public boolean needsUpdate(String instId, long threshold) {
        long lastUpdate = getLastUpdateTime(instId);
        return System.currentTimeMillis() - lastUpdate > threshold;
    }

    /**
     * 检查服务是否正在运行
     */
    public boolean isRunning() {
        return isRunning.get();
    }

    /**
     * 手动添加产品订阅
     */
    public void addSubscription(String instId) {
        if (instId != null && !instId.isEmpty()) {
            subscribedProducts.add(instId);
        }
    }

    /**
     * 手动删除产品订阅
     */
    public void removeSubscription(String instId) {
        if (instId != null && !instId.isEmpty() && !instId.equals(defaultInstId)) {
            subscribedProducts.remove(instId);
        }
    }
}
